using System;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using RxHeat.Common;

namespace RxHeat
{
    /// <summary>
    ///     RxHeat Core Observable helper extension methods
    /// </summary>
    public static class ObservableExtensions
    {
        #region Memory Management Helpers

        /// <summary>
        ///     Add the <see cref="IDisposable" /> object to a Disposable Collection <seealso cref="System.Reactive.Disposables" />.
        ///     Intended as a helper method for Reactive Subscriptions.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="observableSubscription">The observable subscription.</param>
        /// <param name="disposableCollection">The Disposable collection.</param>
        /// <returns>
        ///     Self reference, to allow multiple tracking
        /// </returns>
        public static IDisposable TrackDisposable<T>(this IDisposable observableSubscription, T disposableCollection)
            where T : IDisposable, ICollection<IDisposable>
        {
            IDisposable disposable = observableSubscription;
            disposableCollection.Add(disposable);
            return disposable;
        }

        /// <summary>
        ///     Switches the internal disposable of a <see cref="SerialDisposable" /> .
        /// </summary>
        /// <param name="serialDisposable">The serial disposable.</param>
        /// <param name="oDisposable">The new disposable.</param>
        /// <returns></returns>
        public static IDisposable SwitchAndDispose(this SerialDisposable serialDisposable, IDisposable oDisposable)
        {
            serialDisposable.Disposable = oDisposable;
            return oDisposable;
        }

        /// <summary>
        ///     Dispose of all collection items, and of the collection instance if it implements <see cref="IDisposable" />
        ///     and <paramref name="disposeOfSelf" /> is set to true
        /// </summary>
        /// <typeparam name="T">Collection items type</typeparam>
        /// <param name="disposableCollection">The collection to dispose.</param>
        /// <param name="disposeOfSelf">
        ///     if set to <c>true</c> [dispose of self].
        /// </param>
        public static void DeepDispose<T>(this IEnumerable<T> disposableCollection, bool disposeOfSelf)
            where T : IDisposable
        {
            disposableCollection.ForEach(x => x.Dispose());
            if (disposeOfSelf)
            {
                var dis = (disposableCollection as IDisposable);
                if (dis != null) dis.Dispose();
            }
        }

        /// <summary>
        ///     Dispose of all collection items, then it <see cref="ICollection{T}.Clear" />. At the end it will Dispose of the collection instance if it implements
        ///     <see cref="IDisposable" /> and <paramref name="disposeOfSelf" /> is set to true.
        /// </summary>
        /// <typeparam name="T">Collection items type</typeparam>
        /// <param name="disposableCollection">The collection to dispose.</param>
        /// <param name="disposeOfSelf">
        ///     if set to <c>true</c> [dispose of self].
        /// </param>
        public static void DeepDispose<T>(this ICollection<T> disposableCollection, bool disposeOfSelf = false)
            where T : IDisposable
        {
            if (disposableCollection != null)
            {
                disposableCollection.ForEach(x => x.Dispose());
                disposableCollection.Clear();
                if (disposeOfSelf)
                {
                    var dis = (disposableCollection as IDisposable);
                    if (dis != null) dis.Dispose();
                }
            }
        }

        #endregion

        /// <summary>
        ///     Funnel the source stream through a new <see cref="BehaviorSubject{T}" />.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">The source stream</param>
        /// <returns></returns>
        public static IObservable<T> ToBehaivorSubject<T>(this IObservable<T> source)
        {
            return ToBehaivorSubject(source, default(T));
        }

        /// <summary>
        ///     Funnel the source stream through a new <see cref="BehaviorSubject{T}" />.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">The source.</param>
        /// <param name="initialValue">The initial value.</param>
        /// <returns></returns>
        public static IObservable<T> ToBehaivorSubject<T>(this IObservable<T> source, T initialValue)
        {
            IDisposable sourceDisposable = null;
            var proxy = new BehaviorSubject<T>(initialValue);
            return Observable.Defer(
                () =>
                    {
                        sourceDisposable = source.Subscribe(proxy.OnNext, proxy.OnError, proxy.OnCompleted);
                        return proxy;
                    })
                             .Finally(() =>
                                 {
                                     if (sourceDisposable != null)
                                         sourceDisposable.Dispose();
                                     proxy.Dispose();
                                 });
        }

        /// <summary>
        ///     Tunnels the source stream through a protected ReplaySubject, so that all subscribers are replayed the entire value set.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">The source.</param>
        /// <param name="serialDisposable">The serial disposable.</param>
        /// <returns></returns>
        public static IObservable<T> ToReplayObservable<T>(this IObservable<T> source, SerialDisposable serialDisposable)
        {
            var ret = new ReplaySubject<T>();
            IDisposable subsciptionDisposable = null;
            IDisposable allDisposable = Disposable.Create(() =>
                {
                    if (subsciptionDisposable != null) subsciptionDisposable.Dispose();
                    ret.Dispose();
                });
            serialDisposable.Disposable = allDisposable;
            subsciptionDisposable = source.Subscribe(ret);
            return ret.AsObservable();
        }

        /// <summary>
        ///     To the replay observable.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">The source.</param>
        /// <param name="allDisposable">All disposable.</param>
        /// <returns></returns>
        public static IObservable<T> ToReplayObservable<T>(this IObservable<T> source, out IDisposable allDisposable)
        {
            var ret = new ReplaySubject<T>();
            IDisposable subsciptionDisposable = null;
            allDisposable = Disposable.Create(() =>
                {
                    if (subsciptionDisposable != null) subsciptionDisposable.Dispose();
                    ret.Dispose();
                });
            subsciptionDisposable = source.Subscribe(ret);
            return ret.AsObservable();
        }

        #region Reflection Helpers

        /// <summary>
        ///     Gets the name of the member
        /// </summary>
        /// <typeparam name="TMember"></typeparam>
        /// <param name="propertyExpression"></param>
        /// <returns></returns>
        public static String GetMemberName<TMember>(Expression<Func<TMember>> propertyExpression)
        {
            if (propertyExpression != null)
            {
                var body = propertyExpression.Body as MemberExpression;

                if (body != null)
                    return body.Member.Name;
            }

            return String.Empty;
        }

        #endregion

        #region Operator Helpers

        /// <summary>
        ///     Will return the first value on any of the supplied source streams matching the supplied <see cref="Predicate{T}" />
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">The source.</param>
        /// <param name="predicate">The predicate.</param>
        /// <param name="faultValue">The fault value.</param>
        /// <returns></returns>
        public static IObservable<T> Or<T>(this IObservable<T>[] source, Func<T, bool> predicate,
                                           T faultValue = default(T))
        {
            return Or(source, predicate, null, faultValue);
        }

        /// <summary>
        ///     Will return the first value on any of the supplied source streams matching the supplied <see cref="Predicate{T}" />
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">The source.</param>
        /// <param name="predicate">The predicate.</param>
        /// <param name="scheduler">The scheduler.</param>
        /// <param name="faultValue">The fault value.</param>
        /// <returns></returns>
        public static IObservable<T> Or<T>(this IObservable<T>[] source, Func<T, bool> predicate,
                                           IScheduler scheduler, T faultValue = default(T))
        {
            return Observable.Defer(() => Observable.Create<T>(observer =>
                {
                    string internalId = "X-" + DateTime.Now.Millisecond;
                    Console.WriteLine(" -- {0} Thread ::: {1}", internalId, Thread.CurrentThread.ManagedThreadId);
                    bool atLeastOneWasTrue = false;

                    return
                        source
                            .Select(x => (scheduler != null) ? x.SubscribeOn(ThreadingSettings.Default) : x)
                            .Merge()
                            .Take(source.Length)
                            .Subscribe(
                                x =>
                                    {
                                        if (predicate(x))
                                        {
                                            observer.OnNext(x);
                                            atLeastOneWasTrue = true;
                                            observer.OnCompleted();
                                        }
                                    },
                                observer.OnError, () =>
                                    {
                                        if (!atLeastOneWasTrue)
                                            observer.OnNext(faultValue);
                                        Console.WriteLine("Failed to Find any OR value, OnCompleted");
                                        observer.OnCompleted();
                                    });
                }));
        }

        #endregion
    }
}